
#include "core/service.h"
#include "core/modules.h"
#include "core/threadpool.h"
#include "core/dest_cache.h"
#include "core/core_message.h"
#include "core/core_error.h"
#include "core/api/core_api.h"
#include "x_endian.h"

#define LOG_SVR(v) LOG(v) << "[" << name_ << "] "

class Service::ServiceImpl {
 public:
  ServiceImpl(Service* service, Handle h, const std::string& name,
              const std::string& module_name, const std::string& args) 
      : service_(service),
        context_(NULL),
        name_(name),
        args_(args),
        handle_(h),
        session_(0),
        msg_cbuf_(32),
        executing_(false) {
    module_ptr_ = Modules::Instance().Query(module_name);  
    if (!module_ptr_) {
      THROW_EXTRA(core_error::load_module_failed, module_name);
    }
    context_ = module_ptr_->Create();
  }

  ~ServiceImpl(void) {
    module_ptr_->Release(context_);
  }

  int32_t Init(void) {
    return module_ptr_->Init(service_, args_.c_str());
  }

  void Push(const CoreMessage& msg) {
    boost::mutex::scoped_lock l(mutex_);
    if (msg_cbuf_.full()) 
      msg_cbuf_.set_capacity(msg_cbuf_.capacity() * 2);

    msg_cbuf_.push_back(msg);

    if (!executing_) {
      ThreadPool::Instance().Add(
          boost::bind(&ServiceImpl::ExecuteMessage, this));
      executing_ = true;
    }
  }

  /*!
   * @brief
   *  message execution closure which is added to ThreadPool when there're 
   *  messages in the queue
   * @return 
   *  true if the queue is empty, which means the closure will be removed 
   *  from the ThreadPool
   */
  bool ExecuteMessage(void) {
    if (!msg_cbuf_.empty()) {
      boost::mutex::scoped_lock lock(mutex_);
      if (!msg_cbuf_.empty()) {
        CoreMessage msg = msg_cbuf_.front();
        msg_cbuf_.pop_front();

        executing_ = !msg_cbuf_.empty();
        lock.unlock();

        switch (message_type(&msg)) {
          case kLogic: {
            if (msg_handler_) {
              msg_handler_(service_, msg.source_handle, msg.session, 
                           msg.data, message_length(&msg));
            } else {
              LOG_SVR(WARNING) << "msg handler is not set";
            }
            break;              
          }
          case kTimer: {
            OnTimer(msg);                
            break;
          }
          default: {
            LOG_SVR(WARNING) << "unknown msg type(" 
                             << message_type(&msg) << ")";
          }
        }          

        if (message_is_owner(&msg))
          delete [] msg.data;

        return !executing_;
      }
    }
    return true;
  }

  void OnTimer(const CoreMessage& msg) {
    // get user_data and timer_cb
    uint16_t sz = message_length(&msg);
    if (sz != sizeof(uintptr_t) * 2) {
      LOG_SVR(ERROR) << "invalid msg size(" << sz << ")";
      return;
    }
    const uintptr_t* data = reinterpret_cast<const uintptr_t*>(msg.data); 
    void* user_data = reinterpret_cast<void*>(betoh_ptr(data[0]));
    timer_callback timer_cb = reinterpret_cast<timer_callback>(
        betoh_ptr(data[1]));
    timer_cb(service_, msg.session, user_data);
  }

 public:
  typedef boost::circular_buffer<CoreMessage> MessageQueue;

  Service*       service_;
  void*          context_;
  std::string    name_;       // name of the service
  std::string    args_;
  Handle         handle_;     // handle alloc by handle_pool
  uint32_t       session_;    // net packet session id, monotonously increase
  ModulePtr      module_ptr_;
  MessageQueue   msg_cbuf_;
  MessageHandler msg_handler_;
  bool           executing_;
  boost::mutex   mutex_;
};

#undef LOG_SVR

//////////////////////////////////////////////////////////////////////////////
Service::Service(Handle h, const std::string& name, 
                 const std::string& module_name, const std::string& args)
    : pimpl_(new Service::ServiceImpl(this, h, name, module_name, args)) {}

void Service::SetMsgHandler(const MessageHandler& mh) {
  pimpl_->msg_handler_ = mh;
}

void Service::Push(const CoreMessage& msg) {
  pimpl_->Push(msg);
}

uint32_t Service::AllocSession(void) {
  // avoid -1 when wrap back, because -1 is represent as invalid session value
  uint32_t new_session = pimpl_->session_++;
  return (new_session == -1) ? pimpl_->session_ : new_session;
}

const std::string& Service::name(void) const {
  return pimpl_->name_;
}

const Handle& Service::handle(void) const {
  return pimpl_->handle_;
}

int32_t Service::Init(void) {
  return pimpl_->Init();
}

void* Service::context(void) {
  return pimpl_->context_;
}

//////////////////////////////////////////////////////////////////////////////
namespace {
// name checking
bool IsVariableName(const char* name) {
  if (*name == '_' || isalpha(*name)) {
    for (const char* c = name + 1; *c != '\0'; ++c)
      if (!isalnum(*c) && *c != '_')
        return false;
    return true;
  }
  return false;
}

bool CheckServiceName(const char* name) {
  if (!name) return true;
  if (*name == '.') {
    // local named service
    return strlen(name) <= Config::kMaxServiceNameLength && 
           IsVariableName(name + 1);
  } else if (*name == ':') {
    try {
      boost::lexical_cast<uint32_t>(name + 1);
    } catch (boost::bad_lexical_cast&) {
      return false;
    }
    return true;
  } else {
    // named service
    return strlen(name) <= Config::kMaxServiceNameLength && 
           IsVariableName(name);
  }
}

std::string MakeName(const char* name, Handle h) {
  if (!name)
    return ":" + boost::lexical_cast<std::string>(h.value()); 
  else
    return name;
}

// handle hash function
std::size_t HandleHash(const Handle& h) {
  return boost::hash_value(h.value());
}

}

Services::Services(void) 
    : handle_serial_number_(0),
      handle_map_(31, HandleHash),
      svr_id_(-1) {} // choose a prime as size

void Services ::Init(void) {
  std::string svr_id = Config::Get("server.svr_id");
  svr_id_ = boost::lexical_cast<uint32_t>(svr_id);
}

Handle Services::NextHandle(void) {
  boost::mutex::scoped_lock lock(mutex_);
  return Handle::New(handle_serial_number_++);
}

Handle Services::New(const char* name,
                     const char* module_name,
                     const char* args) {
  if (!CheckServiceName(name)) {
    THROW_EXTRA(core_error::invalid_service_name, name);
  }

  Handle new_handle = NextHandle();
  auto it = handle_map_.find(new_handle);
  if (it != handle_map_.end()) {
    THROW_EXTRA(core_error::dup_service_name, name);
  }

  ServicePtr service_ptr = boost::make_shared<Service>(
      new_handle, 
      MakeName(name, new_handle),
      module_name,
      args ? args : "");
  handle_map_[new_handle] = service_ptr;
  DestCache::Instance().RegisterService(service_ptr->name(), new_handle);

  int32_t init_ret = service_ptr->Init();
  if (init_ret != 0) {
    THROW_EXTRA(core_error::service_init_failed,
                boost::lexical_cast<std::string>(init_ret));
  }

  return new_handle;
}

void Services::Retire(Handle handle) {
  auto it = handle_map_.find(handle);
  if (it != handle_map_.end()) {
    ServicePtr& service_ptr = (*it).second;
    DestCache::Instance().RetireService(service_ptr->name());
    handle_map_.erase(it);
  }
}

void Services::Send(const Handle& destination, const CoreMessage& message) {
  if (svr_id_ == destination.svr_id()) {
    auto it = handle_map_.find(destination);       
    if (it == handle_map_.end()) {
      THROW_EXTRA(core_error::send_dest_not_exist, destination.tostring());
    }
    (*it).second->Push(message);
  } else {
    ServicePtr ejector = ejector_ptr_.lock();
    if (!ejector) {
      auto it = handle_map_.find(DestCache::Instance().Query(".ejector"));
      if (it == handle_map_.end()) {
        THROW(core_error::ejector_is_down);
      }
      ejector = (*it).second;
      ejector_ptr_ = ejector;
    }
    ejector->Push(message);
  }
}

void Services::Send(const char* destination, const CoreMessage& message) {
  Handle h = DestCache::Instance().Query(destination);
  if (!h) {
    THROW_EXTRA(core_error::send_dest_not_exist, destination);
  }
  Send(h, message);
}

